package org.spark

import org.apache.spark.{SparkConf, SparkContext}

object HelloSpark {

  //1.spark 需要新的依赖
  //
  def main(args: Array[String]): Unit = {
    /*  Rdd  是 spark 数据类型 全称 (Resilient Distributed Dataset)弹性分布式数据集
    *     弹性: 计算逻辑，复杂，简单  | 计算存储
    *     分布式: 可以分布式存储  || 分布式 多台机器
    *     数据集: 数据集合比较类似与  list 或者 Array
    *     特点:
    *     1)  1个分区，每个分区是一个数据块
    *     2)  1个依赖的列表  :    rdd前后依赖的关系
    *     3)  1个函数: 技术逻辑
    *     4)  可选的分区器
    *     5)  位置感知调度
    *
    *     Apache  英文翻译过来
    *  */

    //spark rdd 创建  ： 1）  驱动器 Drive（spark context）来 2）
    val conf = new SparkConf().setAppName("helloSpark").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val arr = Array(1,2,3,4,5)
    val rdd = sc.parallelize(arr)
    println(rdd.sum())

    // 控制 日志  ： INFO 信息 ；error 报错 ， fatal 严重 ；warn 警告 ；
    //error         。。。。。Hadoop 报错  没有影响代码运行（ spark 需要一个Hadoop 环境 ）

    //spark 实现 wordcount 词频统计  MR map，reduce ，main
    val word = Array("hello spark","hello scala","hello java","hello 210721")
    val wordRDD = sc.parallelize(word);

    //逻辑
    //rdd  "hello spark","hello scala","hello java","hello 210721"
    //切分（hello,spark）,(hello,scala)...
    //括号  去掉  flatMap   算子  , 切分压平    hello,spark
    //映射  map   (hello,1)(spark,2)...
    //聚合  reduceByKey   (hello,4),(spark,1)...

    //带有  =>  基本都是一个逻辑函数
    //一行代码
    //wordRDD.flatMap((x: String) => { x.split("")}).map((x: String) => { (x, 1) }).reduceByKey((x: Int, y: Int) => { x + y}).foreach(println(_))
    //多行代码
    /*val a = wordRDD.flatMap((x: String) => { x.split("")})
    val b =a.map((x: String) => { (x, 1) })
        b.reduceByKey((x: Int, y: Int) => { x + y}).foreach(println(_))*/

    //现象： 只写了一半的编码 ，输出没有结果。
    //原因： 不是因为没有编码；是因为 没有 遇到一个 action 类型的 算子

    //算子： ，map , flatMap ,reduceByKey ,transformation 转换类型的算子。     算子：计算逻辑
    //action 类型的 算子 ： count() ,sum() , foreach()

    //词频统计    变形
    //wordRDD.flatMap((x: String) => { x.split("")})
    //wordRDD.flatMap((x: String) =>  x.split(""))
    //wordRDD.flatMap((x) =>  x.split(""))
    //wordRDD.flatMap(x =>  x.split(""))
    //wordRDD.flatMap(_.split(""))

    //map((x: String) => { (x, 1) })
    //...

    //reduceByKey((x: Int, y: Int) => { x + y})
    //(x: Int, y: Int) => { x + y}
    //((x: Int, y: Int) =>  x + y
    //((x, y) =>  x + y
    //  _ + _


    /*wordRDD.flatMap(_.split(""))
      .map((_,1) )
      .reduceByKey(_+_)
      .foreach(println(_))*/

    //spark编码 特点： 简洁，优雅，抽象

    //reduceByKey(_+_) // (hello,4) ,(spark,1) .......
    wordRDD.flatMap(_.split("")).map((_,1) ).reduceByKey(_+_).foreach(println(_))

  }

}
